package cn.tianyigps.tianyigpskafka.task.manger;


import cn.tianyigps.tianyigpskafka.task.bean.Task;
import cn.tianyigps.tianyigpskafka.task.util.TaskUtil;
import cn.tianyigps.tianyigpspublic.mongo.dao.BaseDao;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

/**
 * Created with IDEA
 *
 * @description: 生产者
 * @author: cui.guibin
 * @create: 2019/3/18 10:46
 **/
@Component
@Slf4j
public class TaskManger {

    @Resource(name="baseDaoImp")
    private BaseDao<Task> baseDao;
    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * @param: Task task
     * @author cui.guibin
     * @date 2019/3/18 10:50
     * 分发task并存储
     */
    public Map<String, Object> distributeTaskAndSave(Task task){

        Map<String, Object> map = new HashMap<>();
        Boolean aBoolean = TaskUtil.checkDataSize(task,map);
        if (aBoolean) {
            // TODO 不同任务可以选择不存
            if (Optional.ofNullable(task.getNextTask()).isPresent())
                saveTask(task,null);
            try {
                kafkaTemplate.send(task.getTaskType(), TaskUtil.task2Json(task));
            } catch (Exception e) {
                map.put("kafkaSendErr","task数据发送失败");
                e.printStackTrace();
            }
        }else
            log.error("错误！任务数据字节大于1M 或 为空 ：" + task);
        map.put("task",task);
        return map;
    }
    /**
     * @param: Task task
     * @author cui.guibin
     * @date 2019/3/18 10:50
     * 分发task
     */
    public Map<String, Object> distributeTask(Task task){

        Map<String, Object> map = new HashMap<>();
        Boolean aBoolean = TaskUtil.checkDataSize(task,map);
        if (aBoolean) {
            try {
                kafkaTemplate.send(task.getTaskType(), TaskUtil.task2Json(task));
            } catch (Exception e) {
                map.put("kafkaSendErr","task数据发送失败");
                e.printStackTrace();
            }
        }else
            log.error("错误！任务数据字节大于1M 或 为空 ：" + task);
        map.put("task",task);
        return map;
    }

    private void saveTask(Task task, String collectionName) {
        if (StringUtils.isEmpty(collectionName))
            baseDao.insert(task);
        else
            baseDao.insert(task,collectionName);
    }

}
